<?php

namespace App\Support;

use App\Kuafu\Config\Model\ConfigModel;
use App\Kuafu\Config\Service\ConfigService;
use App\Kuafu\Mqtt\Model\MqttShopUserModel;
use App\Kuafu\Nsq\Enum\NsqCodeEnum;
use App\Library\FeiShu\SendFs;
use App\Library\SnowFlake\SnowFlake;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Kuafu\Nsq\NsqPublisher;


class MQTT
{
    // POS 类型
    const TOPIC_TYPE_DEVICE_NUM = 1001;
    // 设备类型
    const TOPIC_TYPE_POS_NUM = 1000;

    const TOPIC_TYPE_POS_NAME = 'shop/%s/pos';
    const TOPIC_TYPE_DEVICE_NAME = 'shop/%s/device';
    const TOPIC_TYPE_APP_WEB = 'shop/%s/app_web';

    const TOPIC_TYPE_SHOP_LIST = [
        self::TOPIC_TYPE_DEVICE_NUM => self::TOPIC_TYPE_DEVICE_NAME,
    ];

    const TOPIC_TYPE_POS_LIST = [
        self::TOPIC_TYPE_POS_NUM => self::TOPIC_TYPE_POS_NAME,
    ];

    const MQTT_CLIENT_CONNECT_POS = 1; // POS
    const MQTT_CLIENT_CONNECT_SHOP = 2; // 门店设备
    const MQTT_CLIENT_CONNECT_APP_WEB = 3; // APP WEB

    const MQTT_CONNECT_TYPE_TEXT = [
        self::MQTT_CLIENT_CONNECT_POS => 'POS',
        self::MQTT_CLIENT_CONNECT_SHOP => '门店设备',
        self::MQTT_CLIENT_CONNECT_APP_WEB => 'APP WEB',
    ];

    public static function initServer($clientId, int $connectType = self::MQTT_CLIENT_CONNECT_POS)
    {
        list($server, $port, $username, $password, $_) = self::getConfigByConnectType($connectType);

        $server = new phpMQTT($server, $port, $clientId);
        if ($server->connect(false, null, $username, $password)) {
            return $server;
        }

        return false;
    }

    public static function makeTopic($shopId, int $connectType = self::MQTT_CLIENT_CONNECT_POS): string
    {
        switch ($connectType) {
            case self::MQTT_CLIENT_CONNECT_APP_WEB:
                return sprintf(self::TOPIC_TYPE_APP_WEB, $shopId);
            case self::MQTT_CLIENT_CONNECT_SHOP:
                return sprintf(self::TOPIC_TYPE_DEVICE_NAME, $shopId);
            case self::MQTT_CLIENT_CONNECT_POS:
            default:
                return sprintf(self::TOPIC_TYPE_POS_NAME, $shopId);
        }
    }

    public static function makeTopicList($shopId, array $topicList = self::TOPIC_TYPE_POS_LIST): array
    {
        $list = [];
        foreach ($topicList as $key => $item) {
            $list[$key] = sprintf($item, $shopId);
        }
        return $list;
    }

    public static function makeUsername($shopId, int $connectType): string
    {
        if (in_array($connectType, [self::MQTT_CLIENT_CONNECT_POS, self::MQTT_CLIENT_CONNECT_SHOP])) {
            return (string)$shopId;
        }

        return $connectType . '_' . $shopId;
    }

    public static function makeClientId($userId = 0): string
    {
        return $userId > 0 ? "user-{$userId}-" . uniqid() : "client-" . uniqid();
    }

    public static function send(
        $shopId,
        $cmd,
        $params,
        int $connectType = self::MQTT_CLIENT_CONNECT_POS
    )
    {
        try {
            $msg = [
                'id' => SnowFlake::nextIdCache(),
                'cmd' => $cmd,
                'params' => $params
            ];

            // 保存到数据库
            \App\Kuafu\Mqtt\Model\Log::add($shopId, $msg);

            // 写入 MQTT 队列
            \App\Kuafu\Mqtt\Model\Queue::add($shopId, $msg);

        } catch (\Exception $exception) {
            Log::error('MQTT_ERROR_1', [
                'line' => $exception->getLine(),
                'msg' => $exception->getMessage(),
            ]);
            SendFs::newRobot()->setTitleContent('MQTT_1 入库报错了 !! ', json_encode($exception))->send();
        }

        $clientId = self::makeClientId();

        try {
            $server = self::initServer($clientId, $connectType);
            if ($server) {
                $server->publish(self::makeTopic($shopId, $connectType), json_encode($msg), 2);
                $server->close();
            } else {
                Log::debug(json_encode($server, '320'));
            }
        } catch (\Exception $exception) {
            Log::error('MQTT_ERROR_2', [
                'line' => $exception->getLine(),
                'msg' => $exception->getMessage(),
            ]);
            SendFs::newRobot()->setTitleContent('MQTT_2 服务异常 !! ', json_encode($exception))->send();
        }
    }

    /**
     * 获取 订单-MQTT 账号
     *
     * @param int $userId 用户 ID
     * @param int $shopId 门店 ID
     * @param false $forceRefresh 强制刷新
     * @return array
     */
    public static function getAccount(int $userId, int $shopId, bool $forceRefresh = false): array
    {
        $access = [
            'host' => config("mqtt.MQTT_HOST"),
            'port' => config("mqtt.MQTT_PORT"),
            'username' => self::makeUsername($shopId, self::MQTT_CLIENT_CONNECT_POS),
            'password' => md5(uniqid()),
            'topic' => self::makeTopic($shopId),
//            'topicList' => self::makeTopicList($shopId),
            'clientId' => self::makeClientId($userId)
        ];

        $db = DB::connection('mysqlMQTT');
        if ($forceRefresh === false) {
            $data = $db->table("mqtt_user")->where("username", $access["username"])->first();
            if ($data) {
                $access['password'] = $data->password;
                return $access;
            }
        }

        try {
            $db->insert("INSERT INTO `mqtt_user` (`username`, `password`) VALUES (?, ?)",
                [$access['username'], $access['password']]);
        } catch (\Exception $e) {
            Log::warning($e);
        }

        return $access;
    }

    /**
     * 获取配置
     * @param int $connectType
     * @return array
     */
    private static function getConfigByConnectType(int $connectType): array
    {
        $wsPort = '';
        switch ($connectType) {

            case self::MQTT_CLIENT_CONNECT_APP_WEB:
                $server = config("mqtt.app_web.mqttShopHost");
                $port = config("mqtt.app_web.mqttShopPort");
                $wsPort = config("mqtt.app_web.mqttShopWsPort");
                $username = config("mqtt.app_web.mqttAdmin");
                $password = config("mqtt.app_web.mqttPassword");
                break;

            case self::MQTT_CLIENT_CONNECT_SHOP:
                $server = config("mqtt.shop.mqttShopHost");
                $port = config("mqtt.shop.mqttShopPort");
                $username = config("mqtt.shop.mqttAdmin");
                $password = config("mqtt.shop.mqttPassword");
                break;
            case self::MQTT_CLIENT_CONNECT_POS:
            default:
                $server = config("mqtt.MQTT_HOST");
                $port = config("mqtt.MQTT_PORT");
                $username = config("mqtt.MQTT_ADMIN");
                $password = config("mqtt.MQTT_PASSWORD");
        }

        return [$server, $port, $username, $password, $wsPort];
    }

    public static function getAccountByConnectType(int $userId, int $shopId, int $connectType, bool $forceRefresh = false): array
    {
        list($host, $port, $_, $_, $wsPort) = self::getConfigByConnectType($connectType);

        $access = [
            'host'     => $host,
            'port'     => $port,
            'wsPort'  => $wsPort,
            'username' => self::makeUsername($shopId, $connectType),
            'password' => md5(uniqid()),
            'topic'    => self::makeTopic($shopId, $connectType),
            'clientId' => self::makeClientId($userId)
        ];

        if ($forceRefresh === false) {
            $data = MqttShopUserModel::query()
                                     ->where('username', $access["username"])
                                     ->where('connect_type', $connectType)
                                     ->first();
            if ($data) {
                $access['password'] = $data->password_hash;
                return $access;
            }
        }

        // 第一次进，如果并发调用接口会导致触发唯一索引报错
        try {
            MqttShopUserModel::query()->insert([
                'connect_type'  => $connectType,
                'username'      => $access['username'],
                'password_hash' => $access['password'],
                'is_superuser'  => 0,
                'created'       => date('Y-m-d H:i:s')
            ]);

        } catch (\Exception $e) {
            Log::warning($e);
        }

        return $access;
    }

    /**
     * mqtt订阅topics
     * @param int $connectType
     * @return void
     */
    public function subscribe(int $connectType): void
    {
        $clientId = self::makeClientId();
        $server   = self::initServer($clientId, $connectType);

        $tps   = json_decode(ConfigModel::getConfigByTypeKeyOne('mqtt', 'mqttReceiveTopics')->value, true);
        $lists = $tps[$connectType] ?? [];
        if (!$lists) {
            Log::warning("MQTT subscribe empty connectType: {$connectType}");
            return;
        }

        $topics = [];
        foreach ($lists as $val) {
            $topics[$val['topic']] = ['qos' => 0, 'function' => [$this, 'subscribeMqttPublishNsq']];
        }

        // demo
        // $topics['bluerhinos/phpMQTT/examples/publishtest/#'] = array('qos' => 0, 'function' => [$this, 'procMsg']);

        $server->subscribe($topics, 0);

        while ($server->proc()) {}

        $server->close();
    }

    /**
     * 订阅mqtt & 发布mqtt
     * @param string $topic
     * @param string $msg
     * @return void
     * @throws \Exception
     */
    public function subscribeMqttPublishNsq(string $topic, string $msg)
    {
        Log::info("subscribeMqttPublishNsq {$topic} - {$msg}");

        if (config('app.env') == 'dev') {
            dump($topic. ' - '. $msg);
        }

        $data = json_decode($msg, true);
        if (!is_array($data)) {
            return;
        }

        (new NsqPublisher(config('nsq.default.address')))
            ->pub(NsqCodeEnum::MQTT_SUBSCRIBE, ['topic' => $topic, 'msg' => $data]);
    }
}
